package org.lwd.datalink;

import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.lwd.datalink.api.probe.ProbeManager;
import org.lwd.datalink.config.properties.WorkerProperties;
import org.lwd.datalink.core.boot.BootMode;
import org.lwd.datalink.core.boot.PluginListenersBootStrap;
import org.lwd.datalink.core.boot.SigarBootStrap;
import org.lwd.datalink.core.runtime.Keeper;
import org.lwd.datalink.core.runtime.TaskConfigManager;
import org.lwd.datalink.core.runtime.TaskPositionManager;
import org.lwd.datalink.core.runtime.TaskStatusManager;
import org.lwd.datalink.core.runtime.TaskSyncStatusManager;
import org.lwd.datalink.core.runtime.Worker;
import org.lwd.datalink.core.runtime.WorkerConfig;
import org.lwd.datalink.core.runtime.WorkerController;
import org.lwd.datalink.core.runtime.coordinate.WorkerKeeper;
import org.lwd.datalink.core.runtime.rest.RestServer;
import org.lwd.datalink.core.runtime.standalone.StandaloneWorkerKeeper;
import org.lwd.datalink.domain.worker.WorkerInfo;
import org.lwd.datalink.errors.DatalinkException;
import org.lwd.datalink.service.SysPropertiesService;
import org.lwd.datalink.service.TaskConfigService;
import org.lwd.datalink.service.WorkerService;
import org.lwd.datalink.utils.IPUtils;
import org.lwd.datalink.utils.spring.SpringUtils;
import org.lwd.datalink.zookeeper.DLinkZkUtils;
import org.lwd.datalink.zookeeper.ZkConfig;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

import java.io.IOException;
import java.net.URI;
import java.util.Map;

/**
 * 启动程序
 *
 * @author linwd
 */
@SpringBootApplication
@Slf4j
public class WorkerApplication {

    public void boot(String[] args) {
        try {
            //initial WorkerConfig
            Map<String, String> workerProps = Maps.newHashMap();

            WorkerProperties workerProperties = SpringUtils.getBean(WorkerProperties.class);
            initProps(workerProps, workerProperties);
            //initial zkclient
            DLinkZkUtils.init(new ZkConfig(
                    workerProperties.getZkServers(),
                    workerProperties.getZkSessionTimeout(),
                    workerProperties.getZkConnectionTimeout()), workerProperties.getZkRoot());

            //初始化数据库中参数
            buildWorkerDBProps(workerProps);

            //init WorkerConfig
            WorkerConfig config = WorkerConfig.fromProps(workerProps, true);

            //get boot mode
            String bootMode = config.getString(WorkerConfig.WORKER_BOOT_MODE_CONFIG);

            //initial plugin EventBus（Listener）
            PluginListenersBootStrap.boot(config);

            SigarBootStrap.boot();

            //initial restserver
            RestServer rest = new RestServer(config);
            URI advertisedUrl = rest.advertisedUrl();
            String restUrl = advertisedUrl.getHost() + ":" + advertisedUrl.getPort();
            String workerId = config.getString(CommonClientConfigs.CLIENT_ID_CONFIG);

            //initial worker
            Time time = new SystemTime();
            Worker worker = new Worker(
                    workerId,
                    time,
                    config,
                    new TaskPositionManager(config),
                    new TaskSyncStatusManager(config),
                    ProbeManager.getInstance()
            );

            //initial Keeper
            Keeper keeper = buildKeeper(
                    config,
                    worker,
                    time,
                    restUrl,
                    bootMode
            );

            //start datalink worker
            log.info("## start the datalink worker.");
            final WorkerController controller = new WorkerController(keeper, rest);
            controller.startup();
            log.info("## the datalink worker is running now ......");
            Runtime.getRuntime().addShutdownHook(new Thread() {

                public void run() {
                    try {
                        log.info("## stop the datalink worker");
                        controller.shutdown();
                    } catch (Throwable e) {
                        log.warn("##something goes wrong when stopping datalink worker", e);
                    } finally {
                        log.info("## datalink worker is down.");
                    }
                }

            });

        } catch (Throwable e) {
            log.error("## Something goes wrong when starting up the datalink worker:", e);
            System.exit(1);
        }
    }

    /**
     * 初始化worker参数
     *
     * @param workerProps
     * @param workerProperties
     */
    private void initProps(Map<String, String> workerProps, WorkerProperties workerProperties) {
        workerProps.put(WorkerConfig.ZK_SERVER_CONFIG, workerProperties.getZkServers());
        workerProps.put(WorkerConfig.ZK_SESSION_TIMEOUT_MS_CONFIG, String.valueOf(workerProperties.getZkSessionTimeout()));
        workerProps.put(WorkerConfig.ZK_CONNECTION_TIMEOUT_MS_CONFIG, String.valueOf(workerProperties.getZkConnectionTimeout()));
        workerProps.put(WorkerConfig.ZK_ROOT_CONFIG, workerProperties.getZkRoot());
        workerProps.put(WorkerConfig.WORKER_BOOT_MODE_CONFIG, workerProperties.getBootMode());
        workerProps.put(WorkerConfig.CLASSLOADER_TYPE_CONFIG, workerProperties.getClassloaderType());
        workerProps.put(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG, workerProperties.getBootstrapServers());
    }

    /**
     * 初始化数据库中参数
     *
     * @param workerProps
     * @throws IOException
     */
    private void buildWorkerDBProps(Map<String, String> workerProps) throws IOException {
        WorkerInfo workerInfo;
        WorkerService service = SpringUtils.getBean(WorkerService.class);
        String clientId = workerProps.get(CommonClientConfigs.CLIENT_ID_CONFIG);
        if (!StringUtils.isBlank(clientId)) {
            workerInfo = service.getById(Long.valueOf(clientId));
        } else {
            workerInfo = service.getWorker(IPUtils.getHostIp());
        }

        if (workerInfo != null) {
            workerProps.put(WorkerConfig.GROUP_ID_CONFIG, String.valueOf(workerInfo.getGroupId()));
            workerProps.put(WorkerConfig.REST_PORT_CONFIG, workerInfo.getRestPort().toString());
            workerProps.put(WorkerConfig.REST_HOST_IP_CONFIG, workerInfo.getWorkerAddress());

            if (StringUtils.isBlank(clientId)) {
                workerProps.put(CommonClientConfigs.CLIENT_ID_CONFIG, String.valueOf(workerInfo.getId()));
            }
            workerProps.put(WorkerConfig.LAB_ID, "1");
        } else {
            throw new DatalinkException(String.format("Worker is not found for client id [%s] or ip [%s] ", clientId, IPUtils.getHostIp()));
        }

        //ddl_sync、sync_auto_addcolumn参数从数据库中读取
        SysPropertiesService propertiesService = SpringUtils.getBean(SysPropertiesService.class);
        Map<String, String> map = propertiesService.map();
        workerProps.put(WorkerConfig.DDL_SYNC_CONFIG, StringUtils.isNotBlank((map.get(WorkerConfig.DDL_SYNC_CONFIG))) ? map.get(WorkerConfig.DDL_SYNC_CONFIG) : String.valueOf(WorkerConfig.DDL_SYNC_DEFAULT));
        workerProps.put(WorkerConfig.SYNC_AUTO_ADD_COLUMN_CONFIG, StringUtils.isNotBlank((map.get(WorkerConfig.SYNC_AUTO_ADD_COLUMN_CONFIG))) ? map.get(WorkerConfig.SYNC_AUTO_ADD_COLUMN_CONFIG) : String.valueOf(WorkerConfig.SYNC_AUTO_ADD_COLUMN_DEFAULT));
    }

    /**
     * Keeper
     *
     * @param config
     * @param worker
     * @param time
     * @param restUrl
     * @param bootMode
     * @return
     */
    private Keeper buildKeeper(WorkerConfig config, Worker worker, Time time, String restUrl, String bootMode) {
        Keeper keeper;
        if (BootMode.DISTRIBUTED.equals(bootMode)) {
            keeper = new WorkerKeeper(
                    config,
                    time,
                    worker,
                    new TaskStatusManager(),
                    new TaskConfigManager(config.getString(WorkerConfig.GROUP_ID_CONFIG), SpringUtils.getBean(TaskConfigService.class)),
                    restUrl
            );
        } else if (BootMode.STANDALONE.equals(bootMode)) {
            keeper = new StandaloneWorkerKeeper(
                    config,
                    worker,
                    time,
                    new TaskConfigManager(config.getString(WorkerConfig.GROUP_ID_CONFIG), SpringUtils.getBean(TaskConfigService.class))
            );
        } else {
            throw new DatalinkException("invalid boot mode : " + bootMode);
        }

        return keeper;
    }

    public static void main(String[] args) {
        SpringApplication.run(WorkerApplication.class, args);

        WorkerApplication application = new WorkerApplication();
        application.boot(args);

        System.out.println("(♥◠‿◠)ﾉﾞ  启动成功   ლ(´ڡ`ლ)ﾞ");
    }
}
